/**
 * kl yy-ds @2021
 */
package com.cxps.server.listener;

import com.alibaba.fastjson.JSON;
import com.cxps.server.product.domain.ProductEntity;
import com.cxps.server.product.domain.ProductIndex;
import com.cxps.server.product.mapper.ProductMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.IOException;

/**
 * 监听MQ，进行消费消息
 *
 * @author YouCai.Liu
 * @since 2021/11/22
 * ...
 */
@Slf4j
@Service
@RocketMQMessageListener(consumerGroup = "product-consumer-group", topic = "product_topic")
public class ProductListener implements RocketMQListener<MessageExt> {


    @Override
    public void onMessage(MessageExt message) {
        String tags = message.getTags();
        String msg = new String(message.getBody());
        String msgId = message.getKeys();

        ProductEntity entity = JSON.parseObject(msg, ProductEntity.class);

        try {
            // 先去数据库查 msgId，如果不存在，执行业务逻辑代码，如果存在，忽略消息

            // 处理消息
            if ("ON".equals(tags)) {
                // 上架数据，将数据同步到ES
                addProduct2ES(msg);
            } else if ("OFF".equals(tags)) {
                // 下架数据，将数据从E删除
                delEsById(msg);
            }

            // 将消息ID存入数据库
            // save db
        } catch (Exception e) {
            log.error("商品消息消费失败：" + e.getMessage(), e);
            throw new RuntimeException("商品消息消费失败：" + e.getMessage(), e);
        }
    }

    @Value("${cxps.yyds.product_index_name}")
    private String indexName;

    @Autowired
    private ProductMapper productMapper;

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    /**
     * 根据ID将数据同步到ES
     *
     * @param id
     */
    private void addProduct2ES(String id) throws IOException {

        // 根据ID查询要同步的商品数据
        ProductEntity entity = productMapper.selectById(id);

        if (null == entity) {
            return;
        }

        // 创建一个 ES索引请求的 对象
        IndexRequest indexRequest = Requests.indexRequest(indexName);

        // 去数据库查询商品数据
        // 索引对象
        ProductIndex index = new ProductIndex();
        BeanUtils.copyProperties(entity, index);

        // 设置文档ID
        indexRequest.id(entity.getId());

        // 将对象转换json
        indexRequest.source(JSON.toJSONString(index), XContentType.JSON);
        indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);

        // 发起请求
        restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);

    }

    /**
     * 根据ID删除ES数据
     *
     * @param id
     */
    private void delEsById(String id) throws IOException {
        DeleteRequest deleteRequest = new DeleteRequest(indexName);
        deleteRequest.id(id);
        restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
    }
}
